package com.csdj.order.mq;

import com.csdj.beans.entity.Order;
import com.csdj.beans.interfaces.goods.IGoodsService;
import com.csdj.beans.interfaces.order.IOrderService;
import com.csdj.common.constant.CommonConstant;
import com.csdj.common.constant.OrderConstant;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;

/**
 * MessageListenerCurrently并行消费消息处理器 会用多个线程去消费同一个消息队列 ， 所以不能保证消息的消费顺序
 * 如果要保证消费消息顺序 请使用MessageListenerOrderly
 */
@Service
public class RocketMsgHandler implements MessageListenerOrderly {

    private Logger log= LoggerFactory.getLogger("msg handler");

    @Resource
    private IOrderService orderService;
    @Resource
    private IGoodsService goodsService;

    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        //消费消息的逻辑， 示例只是打印消息内容
        if(msgs != null){
            log.info("msg size:{}",msgs.size());
            msgs.stream().forEach(msg ->{
                log.info("topic:{} queue:{} data:{}",msg.getTopic(),msg.getQueueId(),new String(msg.getBody()));
            });
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        if(msgs != null){
            try{
                log.info("msg size:{}",msgs.size());
                msgs.stream().forEach(msg ->{
                    if(CommonConstant.MQ_TAGS_ORDER_CHECK.equals(msg.getTags())){
                        //十分钟之后检查订单有没有完成支付， 如果没有则撤消订单
                        String orderIdStr = new String(msg.getBody());
                        Order order = orderService.getById(Integer.parseInt(orderIdStr));
                        if(order != null){
                            if(OrderConstant.ORDER_INIT == order.getStatus()){
                                log.info("orderid:{} 没有支付，回撤",order.getId());
                                //订单还没有支付， 撤消
                                order.setStatus(OrderConstant.ORDER_PAY_FAIL);
                                orderService.update(order);
                                //库存回撤
                                goodsService.updateStorage(order.getProdId(),order.getQuantity()*-1);
                            }
                        }
                    }
                    log.info("topic:{} queueid:{},thread:{} data:{}",msg.getTopic(),msg.getQueueId(),Thread.currentThread().getName(),new String(msg.getBody()));
                });
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
